#include "gtest/gtest.h"

#include "SevenRecordTopic.h"

using namespace stream;

class ReaderWrapperTest : public testing::Test {
public:
    void SetUp() override{
        _7record  = new SevenRecordTopic();
    }
	void TearDown() override{
		delete _7record;
	}
	void _sendGtp(size_t index, const std::string & str) {
		std::cout << index << "--sendGtp--" << str << std::endl;
	};
protected:
    SevenRecordTopic * _7record = nullptr;
};

#include "stream/ReaderWrapper.h"

#include "stream/SequenceReader.h"
TEST_F(ReaderWrapperTest, testSequenceReader){
    auto reader = std::shared_ptr<SequenceReader>(::stream::makeSequenceReader(_7record->topic(), 0ull));
	auto wrapper = std::make_shared<ReaderWrapper<SequenceReader>>(reader);
	ASSERT_TRUE(wrapper->open());

    size_t count  = 0;
	for(auto && p : *wrapper){
        ++ count;
        _sendGtp(p.first, *p.second);
	}

    wrapper->close();
    ASSERT_EQ(count, _7record->expectRecord());
}


#include "stream/Reader.h"
TEST_F(ReaderWrapperTest, testReader){
    auto reader = std::shared_ptr<Reader>(::stream::makeReader(_7record->topic(), 0ull));
	auto wrapper = std::make_shared<ReaderWrapper<Reader>>(reader);
	ASSERT_TRUE(wrapper->open());

	size_t offset = 0;
	for(auto && pstr : *wrapper){
		_sendGtp(offset, *pstr);
		++offset;
	}

	wrapper->close();
    ASSERT_EQ(offset, _7record->expectRecord());
}
